package com.shilei.system_io.v1;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.Channel;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @description: selector组
 * @author: 81271
 * @date: 2020-08-03 18:55
 */
public class SelectorThreadGroup {
    SelectorThread[] sts;
    ServerSocketChannel server = null;
    AtomicInteger xid = new AtomicInteger(0);

    SelectorThreadGroup stg = this;

    public SelectorThreadGroup(int num) {
        //num线程数
        sts = new SelectorThread[num];
        for (int i = 0; i < num; i++) {
            sts[i] = new SelectorThread(this);

            new Thread(sts[i]).start();
        }
    }

    public void setWorker(SelectorThreadGroup worker) {
        this.stg = worker;
    }


    public void bind (int port) {
        try {

            server = ServerSocketChannel.open();
            server.configureBlocking(false);
            server.bind(new InetSocketAddress(port));

            //注册到那个selector上呢？
            nextSelectorV3(server);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void nextSelector(Channel c) {
        ServerSocketChannel server = (ServerSocketChannel) c;
        try {
            SelectorThread next = next();
            System.out.println("wakeup");
            next.selector.wakeup();

            server.register(next.selector, SelectionKey.OP_ACCEPT);
            System.out.println("register");
        } catch (ClosedChannelException e) {
            e.printStackTrace();
        }
    }

    public void nextSelectorV2(Channel c) {
        if (c instanceof ServerSocketChannel) {
            try {
                sts[0].lbq.put(c);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            sts[0].selector.wakeup();
        } else {
            SelectorThread st = nextV2();
            //1.通过队列传递数据消息
            st.lbq.add(c);
            //2.通过打断阻塞，让对应的线程取自己在打断后完成注册
            st.selector.wakeup();
        }

    }

    public void nextSelectorV3(Channel c) {
        if (c instanceof ServerSocketChannel) {
            SelectorThread st = next();
            try {
                st.lbq.put(c); // 阻塞，直到能够有空间插入元素
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            st.setSelectorThreadGroup(this);
            st.selector.wakeup();
        } else {
            SelectorThread st = nextV3();
            //1.通过队列传递数据消息
            st.lbq.add(c);
            //2.通过打断阻塞，让对应的线程取自己在打断后完成注册
            st.selector.wakeup();
        }

    }


    private SelectorThread next() {
        int index = xid.incrementAndGet() % sts.length;
        return sts[index];
    }

    private SelectorThread nextV2() {
        int index = xid.incrementAndGet() % (sts.length - 1);
        return sts[index + 1];
    }

    private SelectorThread nextV3() {
        int index = xid.incrementAndGet() % stg.sts.length;
        return stg.sts[index];
    }
}
